package thread;

import java.io.IOException;
import java.net.SocketTimeoutException;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.ConcurrentModificationException;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cache.NioSelectorPool;
import cache.SynchronizedQueue;
import cache.SynchronizedStack;
import cachedata.NioChannel;
import cachedata.NioSocketWrapper;
import config.SocketProperties;
import enumerate.SocketEvent;
import util.ExceptionUtils;


public class Poller implements Runnable {
	
	private static final Logger log = LoggerFactory.getLogger(Poller.class);

	private static final long selectorTimeout = 50;

    private Selector selector;
    private final SynchronizedQueue<PollerEvent> events =
            new SynchronizedQueue<>();
    
    private final SynchronizedStack<PollerEvent> eventCache;
    
    private final SynchronizedStack<ProccessWorker> processorCache;

    private volatile boolean close = false;
    private long nextExpiration = 0;//optimize expiration handling

    private AtomicLong wakeupCounter = new AtomicLong(0);

    private volatile int keyCount = 0;
    
    private final NioSelectorPool pool;

    private final ExecutorService executor;
    
    public Poller(NioSelectorPool selectorPool, ExecutorService executor, SynchronizedStack<PollerEvent> eventCache, SynchronizedStack<ProccessWorker> processorCache) {
    	this.pool = selectorPool;
    	this.executor = executor;
    	this.eventCache = eventCache;
    	this.processorCache = processorCache;
       init();
    }
    
    private void init(){
    	 try {
			this.selector = Selector.open();
		} catch (IOException e) {
			e.printStackTrace();
		}
    }

    public int getKeyCount() { return keyCount; }

    public Selector getSelector() { return selector;}

    /**
     * Destroy the poller.
     */
    protected void destroy() {
        // Wait for polltime before doing anything, so that the poller threads
        // exit, otherwise parallel closure of sockets which are still
        // in the poller can cause problems
        close = true;
        selector.wakeup();
    }

    private void addEvent(PollerEvent event) {
        events.offer(event);
        if ( wakeupCounter.incrementAndGet() == 0 ) selector.wakeup();
    }

    /**
     * Add specified socket and associated pool to the poller. The socket will
     * be added to a temporary array, and polled first after a maximum amount
     * of time equal to pollTime (in most cases, latency will be much lower,
     * however).
     *
     * @param socket to add to the poller
     * @param interestOps Operations for which to register this socket with
     *                    the Poller
     */
    public void add(final NioChannel socket, final int interestOps) {
        PollerEvent r = eventCache.pop();
        if ( r==null) r = new PollerEvent(socket,null,interestOps);
        else r.reset(socket,null,interestOps);
        addEvent(r);
        if (close) {
            NioSocketWrapper ka = (NioSocketWrapper)socket.getAttachment();
            processSocket(ka, SocketEvent.STOP, false);
        }
    }
    
    /**
     * Process the given SocketWrapper with the given status. Used to trigger
     * processing as if the Poller (for those endpoints that have one)
     * selected the socket.
     *
     * @param socketWrapper The socket wrapper to process
     * @param event         The socket event to be processed
     * @param dispatch      Should the processing be performed on a new
     *                          container thread
     *
     * @return if processing was triggered successfully
     */
    public boolean processSocket(NioSocketWrapper socketWrapper,
            SocketEvent event, boolean dispatch) {
        try {
            if (socketWrapper == null) {
                return false;
            }
            
           
			ProccessWorker sc = processorCache.pop();
            if (sc == null) {
                sc = createSocketProcessor(socketWrapper, event);
            } else {
                sc.reset(socketWrapper, event);
            }
            if (dispatch && executor != null) {
                executor.execute(sc);
            } else {
                sc.run();
            }
        } catch (RejectedExecutionException ree) {
//           log.warn(sm.getString("endpoint.executor.fail", socketWrapper) , ree);
            return false;
        } catch (Throwable t) {
            ExceptionUtils.handleThrowable(t);
            // This means we got an OOM or similar creating a thread, or that
            // the pool and its queue are full
//           log.error(sm.getString("endpoint.process.fail"), t);
            return false;
        }
        return true;
    }

	private ProccessWorker createSocketProcessor(NioSocketWrapper socketWrapper, SocketEvent event) {
		return new ProccessWorker(socketWrapper, event);
	}

	/**
     * Processes events in the event queue of the Poller.
     *
     * @return <code>true</code> if some events were processed,
     *   <code>false</code> if queue was empty
     */
    public boolean events() {
        boolean result = false;

        PollerEvent pe = null;
        while ( (pe = events.poll()) != null ) {
            result = true;
            try {
                pe.run();
                pe.reset();
                eventCache.push(pe);
            } catch ( Throwable x ) {
                log.error("",x);
            }
        }

        return result;
    }

    /**
     * Registers a newly created socket with the poller.
     *
     * @param socket    The newly created socket
     */
    public void register(final NioChannel socket) {
        socket.setPoller(this);
        NioSocketWrapper ka = new NioSocketWrapper(socket, pool);
        socket.setSocketWrapper(ka);
        ka.setPoller(this);
        ka.setReadTimeout(SocketProperties.getInstance().getSoTimeout());
        ka.setWriteTimeout(SocketProperties.getInstance().getSoTimeout());
//        ka.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
        ka.setReadTimeout(SocketProperties.getInstance().getSoTimeout());
        ka.setWriteTimeout(SocketProperties.getInstance().getSoTimeout());
        ka.interestOps(SelectionKey.OP_READ);//this is what OP_REGISTER turns into.
        PollerEvent r = eventCache.pop();
        if ( r==null) r = new PollerEvent(socket,ka,PollerEvent.OP_REGISTER);
        else r.reset(socket,ka,PollerEvent.OP_REGISTER);
        log.debug(" new PollerEvent");
        addEvent(r);
    }

    public NioSocketWrapper cancelledKey(SelectionKey key) {
        NioSocketWrapper ka = null;
        try {
            if ( key == null ) return null;//nothing to do
            ka = (NioSocketWrapper) key.attach(null);
            if (ka != null) {
                // If attachment is non-null then there may be a current
                // connection with an associated processor.
                //TODO getHandler().release(ka);
            }
            if (key.isValid()) key.cancel();
            if (key.channel().isOpen()) {
                try {
                    key.channel().close();
                } catch (Exception e) {
//                    if (log.isDebugEnabled()) {
//                        log.debug(sm.getString(
//                                "endpoint.debug.channelCloseFail"), e);
//                    }
                }
            }
            try {
                if (ka!=null) {
                    ka.getSocket().close(true);
                }
            } catch (Exception e){
//                if (log.isDebugEnabled()) {
//                    log.debug(sm.getString(
//                            "endpoint.debug.socketCloseFail"), e);
//                }
            }
            if (ka != null) {
                Acceptor.countDownConnection();
            }
        } catch (Throwable e) {
            ExceptionUtils.handleThrowable(e);
            if (log.isDebugEnabled()) log.error("",e);
        }
        return ka;
    }

    /**
     * The background thread that adds sockets to the Poller, checks the
     * poller for triggered events and hands the associated socket off to an
     * appropriate processor as events occur.
     */
    @Override
    public void run() {
        // Loop until destroy() is called
        while (true) {

            boolean hasEvents = false;

            try {
                if (!close) {
                    hasEvents = events();
                    if (wakeupCounter.getAndSet(-1) > 0) {
                        //if we are here, means we have other stuff to do
                        //do a non blocking select
                        keyCount = selector.selectNow();
                    } else {
                        keyCount = selector.select(selectorTimeout);
                    }
                    wakeupCounter.set(0);
                }
                if (close) {
                    events();
                    timeout(0, false);
                    try {
                        selector.close();
                    } catch (IOException ioe) {
//                        log.error(sm.getString("endpoint.nio.selectorCloseFail"), ioe);
                    }
                    break;
                }
            } catch (Throwable x) {
                ExceptionUtils.handleThrowable(x);
                log.error("",x);
                continue;
            }
            //either we timed out or we woke up, process events first
            if ( keyCount == 0 ) hasEvents = (hasEvents | events());

            Iterator<SelectionKey> iterator =
                keyCount > 0 ? selector.selectedKeys().iterator() : null;
            // Walk through the collection of ready keys and dispatch
            // any active event.
            while (iterator != null && iterator.hasNext()) {
                SelectionKey sk = iterator.next();
               
                NioSocketWrapper attachment = (NioSocketWrapper)sk.attachment();
                log.debug("attachment " + attachment);
                // Attachment may be null if another thread has called
                // cancelledKey()
                if (attachment == null) {
                    iterator.remove();
                } else {
                    iterator.remove();
                    processKey(sk, attachment);
                }
            }//while

            //process timeouts
            timeout(keyCount,hasEvents);
        }//while

//        stopLatch.countDown();
    }

    protected void processKey(SelectionKey sk, NioSocketWrapper attachment) {
        try {
            if ( close ) {
                cancelledKey(sk);
            } else if ( sk.isValid() && attachment != null ) {
                if (sk.isReadable() || sk.isWritable() ) {             
                    unreg(sk, attachment, sk.readyOps());
                    boolean closeSocket = false;
                    // Read goes before write
                    if (sk.isReadable()) {
                    	log.debug("isReadable ");
                        if (!processSocket(attachment, SocketEvent.OPEN_READ, true)) {
                            closeSocket = true;
                        }
                    }
                    if (!closeSocket && sk.isWritable()) {
                    	log.debug("isWritable ");
                        if (!processSocket(attachment, SocketEvent.OPEN_WRITE, true)) {
                            closeSocket = true;
                        }
                    }
                    if (closeSocket) {
                        cancelledKey(sk);
                    }
                }
            } else {
                //invalid key
                cancelledKey(sk);
            }
        } catch ( CancelledKeyException ckx ) {
            cancelledKey(sk);
        } catch (Throwable t) {
            ExceptionUtils.handleThrowable(t);
            log.error("",t);
        }
    }


    protected void unreg(SelectionKey sk, NioSocketWrapper attachment, int readyOps) {
        //this is a must, so that we don't have multiple threads messing with the socket
        reg(sk,attachment,sk.interestOps()& (~readyOps));
    }

    protected void reg(SelectionKey sk, NioSocketWrapper attachment, int intops) {
        sk.interestOps(intops);
        attachment.interestOps(intops);
    }

    protected void timeout(int keyCount, boolean hasEvents) {
        long now = System.currentTimeMillis();
        // This method is called on every loop of the Poller. Don't process
        // timeouts on every loop of the Poller since that would create too
        // much load and timeouts can afford to wait a few seconds.
        // However, do process timeouts if any of the following are true:
        // - the selector simply timed out (suggests there isn't much load)
        // - the nextExpiration time has passed
        // - the server socket is being closed
        if (nextExpiration > 0 && (keyCount > 0 || hasEvents) && (now < nextExpiration) && !close) {
            return;
        }
        //timeout
        int keycount = 0;
        try {
            for (SelectionKey key : selector.keys()) {
                keycount++;
                try {
                    NioSocketWrapper ka = (NioSocketWrapper) key.attachment();
                    if ( ka == null ) {
                        cancelledKey(key); //we don't support any keys without attachments
                    } else if (close) {
                        key.interestOps(0);
                        ka.interestOps(0); //avoid duplicate stop calls
                        processKey(key,ka);
                    } else if ((ka.interestOps()&SelectionKey.OP_READ) == SelectionKey.OP_READ ||
                              (ka.interestOps()&SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE) {
                        boolean isTimedOut = false;
                        // Check for read timeout
                        if ((ka.interestOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) {
                            long delta = now - ka.getLastRead();
                            long timeout = ka.getReadTimeout();
                            isTimedOut = timeout > 0 && delta > timeout;
                        }
                        // Check for write timeout
                        if (!isTimedOut && (ka.interestOps() & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE) {
                            long delta = now - ka.getLastWrite();
                            long timeout = ka.getWriteTimeout();
                            isTimedOut = timeout > 0 && delta > timeout;
                        }
                        if (isTimedOut) {
                            key.interestOps(0);
                            ka.interestOps(0); //avoid duplicate timeout calls
                            ka.setError(new SocketTimeoutException());
                            if (!processSocket(ka, SocketEvent.ERROR, true)) {
                                cancelledKey(key);
                            }
                        }
                    }
                }catch ( CancelledKeyException ckx ) {
                    cancelledKey(key);
                }
            }//for
        } catch (ConcurrentModificationException cme) {
            // See https://bz.apache.org/bugzilla/show_bug.cgi?id=57943
//            log.warn(sm.getString("endpoint.nio.timeoutCme"), cme);
        }
        long prevExp = nextExpiration; //for logging purposes only
        nextExpiration = System.currentTimeMillis() +
                SocketProperties.getInstance().getTimeoutInterval();
        if (log.isTraceEnabled()) {
            log.trace("timeout completed: keys processed=" + keycount +
                    "; now=" + now + "; nextExpiration=" + prevExp +
                    "; keyCount=" + keyCount + "; hasEvents=" + hasEvents +
                    "; eval=" + ((now < prevExp) && (keyCount>0 || hasEvents) && (!close) ));
        }

    }

}
